fix(core): restore event emitter for detached tool calls - #2483

Review — Approved ✅

Fix: Restore AgentEventEmitter in Reactor Context when executeToolCalls(...) creates an internal subscription that loses the original streaming context.

Analysis

The root cause is clear: streamEvents() puts the AgentEventEmitter into Reactor Context at subscription time, but executeToolCalls(...) creates a new internal subscription. The contextWrite(ctx -> ctx.putAll(parentCtx)) merges the parent context, but if the parent context itself lost the emitter key (e.g., through a middleware boundary or context isolation), the child agent spawned in AgentSpawnTool#execLocalSync(...) can't find the parent emitter and falls back to non-streaming execution.

The fix adds a fallback in the contextWrite lambda:

1. Merge parent context as before
2. If both SubagentEventBus.CONTEXT_KEY and AgentEventEmitter.CONTEXT_KEY are missing:
   • Prefer eventSink (local streaming subscription) → wrap as AgentEventEmitter
   • Fall back to externalEventEmitter (parent agent forwarding)
3. If either key already exists, preserve it (no override)

This is correct because:

• eventSink is bound to the current CallExecution during streamEvents() setup, so it's always available when streaming
• externalEventEmitter handles the case where this agent was itself called by a parent agent's forwarding
• The guard !merged.hasKey(...) ensures backward compatibility with stream() (deprecated) and explicit event forwarding paths

Test Quality

The test uses a StripToolEventContextMiddleware to simulate the exact failure mode — stripping AgentEventEmitter.CONTEXT_KEY before tool execution. This is a clever and precise test design that:

• Verifies the fix restores the emitter from eventSink
• Confirms AGENT_START / AGENT_END events are still emitted
• Validates exactly 1 ToolResultEndEvent (no duplicates)
• Tests both the "eventSink available" and "externalEventEmitter fallback" paths

Minor Notes

• The lambda cast (AgentEventEmitter) eventSink::next works because AgentEventEmitter is a functional interface. Consider adding a static factory method like AgentEventEmitter.fromSink(FluxSink) for clarity if this pattern appears elsewhere.
• Thread safety: FluxSink.next() is thread-safe per Reactor docs, so concurrent tool calls won't corrupt the event stream.

Verdict: Clean, minimal, well-tested fix for a real Context propagation gap. LGTM.

🤖 AI Review

🤖 AI Review

🤖 AI Review